package drds.global_transaction.api.rpc.rpc_client;

import java.net.InetSocketAddress;

import org.apache.commons.pool.KeyedPoolableObjectFactory;

import drds.common.Log;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.RegisterResourceManagerResponseMessage;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.RegisterTransactionManagerResponseMessage;
import drds.global_transaction.common.exception.FrameworkException;
import drds.global_transaction.common.util.Nets;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import lombok.extern.slf4j.Slf4j;

/**
 * 根据地址连接服务器进行TransactionManager注册
 */
@Slf4j
public class AddressToChannelFactory implements KeyedPoolableObjectFactory<Key, Channel>
{
	private final RpcClient rpcClient;

	public AddressToChannelFactory(RpcClient rpcClient)
	{
		this.rpcClient = rpcClient;
		this.rpcClient.setChannelHandlers(new ChannelHandler[] { rpcClient });
		this.rpcClient.start();
	}

	/**
	 * 输入参数是ky
	 */
	@Override
	public Channel makeObject(Key key) throws Exception
	{
		InetSocketAddress inetSocketAddress = Nets.toInetSocketAddress(key.getAddress());
		Channel channel = rpcClient.connect(inetSocketAddress);
		Object response = null;
		Channel needReturnChannel = null;
		if (null == key.getMessage())
		{
			throw new FrameworkException("register msg is null, role:" + key.getRole().name());
		}
		try
		{
			response = rpcClient.sendAsynchronousRequestWithResponse(null, channel, key.getMessage());
			if (!isRegisterTransactionManagerSuccess(response, key.getRole()))
			{
				rpcClient.onRegisterMessageFail(key.getAddress(), channel, key.getMessage(), response);
				log.info("服务器端版本:" + getVersion(response, key.getRole()));
			} else
			{
				needReturnChannel = channel;
				rpcClient.onRegisterMessageSuccess(key.getAddress(), channel, key.getMessage(), response);
			}
		} catch (Exception e)
		{
			if (channel != null)
			{
				channel.close();
			}
			Log.error(log, Thread.currentThread(), "", e);
			throw new FrameworkException("register error,role:" + key.getRole().name() + ",err:" + e.getMessage());
		}
		return needReturnChannel;
	}

	/**
	 * 注册成功?,这个是带有业务的消息
	 */
	private boolean isRegisterTransactionManagerSuccess(Object response, Role role)
	{
		if (null == response)
		{
			return false;
		}
		if (role.equals(Role.transaction_manager_client))
		{
			if (!(response instanceof RegisterTransactionManagerResponseMessage))
			{
				return false;
			}
			if (((RegisterTransactionManagerResponseMessage) response).isIdentified())
			{
				return true;
			}
		} else if (role.equals(Role.resource_manager_client))
		{
			if (!(response instanceof RegisterResourceManagerResponseMessage))
			{
				return false;
			}
			if (((RegisterResourceManagerResponseMessage) response).isIdentified())
			{
				return true;
			}
		}
		return false;
	}

	private String getVersion(Object response, Role role)
	{
		if (role.equals(Role.transaction_manager_client))
		{
			return ((RegisterTransactionManagerResponseMessage) response).getVersion();
		} else
		{
			return ((RegisterResourceManagerResponseMessage) response).getVersion();
		}
	}

	@Override
	public boolean validateObject(Key key, Channel channel)
	{
		if (null != channel && channel.isActive())
		{
			return true;
		} else
		{
			return false;
		}
	}

	@Override
	public void destroyObject(Key key, Channel channel) throws Exception
	{

		if (null != channel)
		{
			channel.disconnect();
			channel.close();
		}
	}

	@Override
	public void activateObject(Key key, Channel obj) throws Exception
	{
	}

	@Override
	public void passivateObject(Key key, Channel obj) throws Exception
	{

	}
}
